package com.xxd.dfs.namenode;

import com.xxd.dfs.namenode.constant.HeartbeatResponseConstant;
import com.xxd.dfs.namenode.constant.MkdirResponseConstant;
import com.xxd.dfs.namenode.constant.RegisterResponseConstant;
import com.xxd.dfs.namenode.rpc.model.*;
import com.xxd.dfs.namenode.rpc.service.NameNodeServiceGrpc;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-08-14 09:15
 * @description:
 */
public class NameNodeRpcServer implements NameNodeServiceGrpc.NameNodeService {

    private final Logger logger = LoggerFactory.getLogger(NameNodeRpcServer.class);

    private FSNameSystem nameSystem;

    public NameNodeRpcServer(FSNameSystem nameSystem) {
        this.nameSystem = nameSystem;
    }

    /**
     * datanode 发起的 register 注册请求
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void register(RegisterRequest request,
                         StreamObserver<RegisterResponse> responseObserver) {
        try {
            boolean result = nameSystem.register(request.getIp(),
                    request.getHostname(), request.getNioPort());

            RegisterResponse response = RegisterResponse.newBuilder()
                    .setStatus(result ? RegisterResponseConstant.STATUS_SUCCESS :
                            RegisterResponseConstant.STATUS_FAILURE)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            logger.error("{} 注册失败", request.getHostname());
        }
    }

    /**
     * datanode 发起的 heartbeat 请求
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void heartbeat(HeartbeatRequest request,
                          StreamObserver<HeartbeatResponse> responseObserver) {
        try {
            boolean result = nameSystem.heartbeat(request.getIp(), request.getHostname());

            HeartbeatResponse response = HeartbeatResponse.newBuilder()
                    .setStatus(result
                            ? HeartbeatResponseConstant.STATUS_SUCCESS
                            : HeartbeatResponseConstant.STATUS_FAILURE)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            logger.error("心跳发送失败", request.getHostname());
        }
    }

    /**
     * 创建文件目录
     *
     * @param request          绝对路劲
     * @param responseObserver
     */
    @Override
    public void mkdir(MkdirRequest request, StreamObserver<MkdirResponse> responseObserver) {
        logger.info("接收到创建文件的请求，文件绝对路劲为 {}", request.getPath());
        MkdirResponse response;
        try {
            nameSystem.mkdir(request.getPath());
            response = MkdirResponse.newBuilder()
                    .setStatus(MkdirResponseConstant.STATUS_SUCCESS)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            response = MkdirResponse.newBuilder()
                    .setStatus(MkdirResponseConstant.STATUS_FAILURE)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
            logger.error("创建失败", e);
        }
    }

    /**
     * 拉取edit log
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void fetchEditLogs(FetchEditLogRequest request,
                              StreamObserver<FetchEditLogResponse> responseObserver) {
        try {
            FetchEditLogResponse response = nameSystem.fetchEditLogs(request.getSeq());
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            logger.error("拉取失败");
        }
    }

    /**
     * 更新checkpoint Txid
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void updateCheckpointTxid(UpdateCheckpointTxidRequest request,
                                     StreamObserver<UpdateCheckpointTxidResponse> responseObserver) {
        try {
            nameSystem.updateCheckpointTxid(request.getTxid());

            UpdateCheckpointTxidResponse response = UpdateCheckpointTxidResponse.newBuilder()
                    .setStatus(1)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {

        }
    }

    /**
     * 优雅关闭接口
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void shutdown(ShutdownRequest request,
                         StreamObserver<ShutdownResponse> responseObserver) {
        try {
            nameSystem.shutdown();
        } catch (Exception e) {

        }
    }

    /**
     * 创建文件请求
     *
     * @param request
     * @param responseObserver
     */
    @Override
    public void create(CreateFileRequest request,
                       StreamObserver<CreateFileResponse> responseObserver) {
        CreateFileResponse response = null;
        try {
            boolean result = nameSystem.create(request.getFilename());
            response = CreateFileResponse.newBuilder()
                    .setStatus(result ? 1 : 0)
                    .build();

        } catch (Exception e) {
            logger.error("上传文件失败", e);
            response = CreateFileResponse.newBuilder()
                    .setStatus(0)
                    .build();
        }

        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     * 请求分配datanode
     * @param request
     * @param responseObserver
     */
    @Override
    public void allocateDataNodes(AllocateDataNodesRequest request,
                                  StreamObserver<AllocateDataNodesResponse> responseObserver) {
        try {
            String datanodesJson = nameSystem.allocateDataNodes(request.getFilename(), request.getFileSize());

            AllocateDataNodesResponse response = AllocateDataNodesResponse.newBuilder()
                    .setDatanodes(datanodesJson)
                    .build();

            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } catch (Exception e) {
            logger.error("请求分配datanode 失败", e);
        }
    }

    /**
     * datanode接收到block通知请求
     * @param request
     * @param responseObserver
     */
    @Override
    public void informBlockReceived(InformBlockReceivedRequest request,
                                    StreamObserver<InformBlockReceivedResponse> responseObserver) {

    }
}
